#include "../include/ReactorDispatcher.h"

#include "Poco/Random.h"
#include "Poco/Hash.h"

#include "../include/SubReactor.h"

namespace Reactor
{
struct ReactorDispatcherPrivate
{
	ReactorDispatcher::DispatcherPtr _dispatcher;
	const std::vector<std::shared_ptr<SubReactor>>& _subReactors;
	ReactorDispatcherPrivate(const std::vector<std::shared_ptr<SubReactor>>& _subReactors)
		:_subReactors(_subReactors)
	{

	}
};

class MOUDLE_API RoundRobinDispatcher : public ReactorDispatcher::Dispatcher
{
public:
	RoundRobinDispatcher(const std::vector<std::shared_ptr<SubReactor>>& _subReactors)
		:Dispatcher(_subReactors)
	{
		_nextReactorIndex = 0;
	}

	std::shared_ptr<SubReactor> dispatch(const Poco::Net::StreamSocket* _pSocket) override
	{
		if (_pSocket == nullptr || _subReactors.empty())
		{
			return nullptr;
		}
		_nextReactorIndex = _nextReactorIndex % _subReactors.size();
		return _subReactors[_nextReactorIndex++];
	}

private:
	Poco::AtomicCounter _nextReactorIndex{ 0 };
};

class MOUDLE_API RandomDispatcher : public ReactorDispatcher::Dispatcher
{
public:
	RandomDispatcher(const std::vector<std::shared_ptr<SubReactor>>& _subReactors)
		:Dispatcher(_subReactors)
	{
		_rnd.seed();
	}

	std::shared_ptr<SubReactor> dispatch(const Poco::Net::StreamSocket* _pSocket) override
	{
		if (_pSocket == nullptr || _subReactors.empty())
		{
			return nullptr;
		}
		return _subReactors[_rnd.next(static_cast<Poco::UInt32>(_subReactors.size()))];
	}

private:
	Poco::Random _rnd;
};

class MOUDLE_API LeastConnectionDispatcher : public ReactorDispatcher::Dispatcher
{
public:
	LeastConnectionDispatcher(const std::vector<std::shared_ptr<SubReactor>>& _subReactors)
		:Dispatcher(_subReactors)
	{
	}

	Poco::UInt32 subReactorSockets(std::shared_ptr<SubReactor> _reactor)
	{
		if (!_reactor)
		{
			return 0;
		}
		return dynamic_cast<Reactor::SubReactor*>(_reactor.get())->socketCount();
	}

	std::shared_ptr<SubReactor> dispatch(const Poco::Net::StreamSocket* _pSocket) override
	{
		if (_pSocket == nullptr || _subReactors.empty())
		{
			return nullptr;
		}

		int minConnections = subReactorSockets(_subReactors[0]);
		std::shared_ptr<SubReactor> minReactor = _subReactors[0];
		for (size_t i = 1; i < _subReactors.size(); ++i)
		{
			int connections = subReactorSockets(_subReactors[i]);
			if (connections < minConnections)
			{
				minConnections = connections;
				minReactor = _subReactors[i];
			}
		}
		return minReactor;
	}
};

class MOUDLE_API HashDispatcher : public ReactorDispatcher::Dispatcher
{
public:
	HashDispatcher(const std::vector<std::shared_ptr<SubReactor>>& _subReactors)
		:Dispatcher(_subReactors)
	{
	}

	std::shared_ptr<SubReactor> dispatch(const Poco::Net::StreamSocket* _pSocket) override
	{
		if (_pSocket == nullptr || _subReactors.empty())
		{
			return nullptr;
		}
		int _reactorIndex = static_cast<int>(Poco::hash(_pSocket->peerAddress().toString()) % _subReactors.size());
		return _subReactors[_reactorIndex];
	}
};

ReactorDispatcher::ReactorDispatcher(const std::vector<std::shared_ptr<SubReactor>>& _subReactors)
	:d(new ReactorDispatcherPrivate(_subReactors))
{

}

ReactorDispatcher::~ReactorDispatcher()
{

}

void ReactorDispatcher::setDispatcher(DispatcherPtr _dispatcher)
{
	if (!_dispatcher)
	{
		return;
	}
	d->_dispatcher = _dispatcher;
}

ReactorDispatcher::DispatchStrategyFactory ReactorDispatcher::dispatchStrategyFactory()
{
	return DispatchStrategyFactory(d->_subReactors);
}

std::shared_ptr<SubReactor> ReactorDispatcher::reactor(const Poco::Net::StreamSocket* _pSocket)
{
	if (_pSocket == nullptr)
	{
		return nullptr;
	}
	return d->_dispatcher->dispatch(_pSocket);
}

ReactorDispatcher::Dispatcher::Dispatcher(const std::vector<std::shared_ptr<SubReactor>>& _subReactors)
	:_subReactors(_subReactors)
{

}

ReactorDispatcher::DispatchStrategyFactory::DispatchStrategyFactory(const std::vector<std::shared_ptr<SubReactor>>& _subReactors)
	:_subReactors(_subReactors)
{
}

ReactorDispatcher::DispatcherPtr ReactorDispatcher::DispatchStrategyFactory::dispatcher(DispatchStrategy _strategy)
{
	switch (_strategy)
	{
		case Reactor::ReactorDispatcher::DispatchStrategy::RoundRobin:
		{
			return  std::make_shared<RoundRobinDispatcher>(_subReactors);
		}
		break;

		case Reactor::ReactorDispatcher::DispatchStrategy::Random:
		{
			return std::make_shared<RandomDispatcher>(_subReactors);
		}
		break;

		case Reactor::ReactorDispatcher::DispatchStrategy::LeastConnection:
		{
			return std::make_shared<LeastConnectionDispatcher>(_subReactors);

		}
		break;

		case Reactor::ReactorDispatcher::DispatchStrategy::Hash:
		{
			return std::make_shared<HashDispatcher>(_subReactors);
		}
		break;

		default:
			return 0;
			break;
	}
}
}